package way.learning.zz;

import way.learning.zz.event.AfterPollingCycleEvent;
import way.learning.zz.event.BeforePollingCycleEvent;
import way.learning.zz.listener.Rdp4jListener;
import way.learning.zz.source.PolledDirectory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 *
 * 这是根据目录轮询器的设置每n毫秒执行的定时器线程。
 *
 * 它调查相关目录，并通知监听器是否添加/删除/修改了文件，或者是否已经提出/停止了IO错误。
 */
final class ScheduledRunnable implements Runnable {

    private static final Logger LOG = LoggerFactory.getLogger(ScheduledRunnable.class);
    private final DirectoryPoller dp;

    private final Set<Poller> pollers = new CopyOnWriteArraySet<Poller>();

    private final ListenerNotifier notifier;

    /**
     * ExecutorService 是 Java中对线程池定义的一个接口，它java.util.concurrent包中，在这个接口中定义了和后台任务执行相关的方法
     *
     * ExecutorService 实际上是一个线程池的管理工具
     */
    private final ExecutorService executor;

    //构造函数
    ScheduledRunnable(DirectoryPoller directoryPoller) {
        dp = directoryPoller;
        this.notifier = dp.notifier;

        //为每一个轮询目录构建一个poller线程
        for (PolledDirectory directory : dp.directories) {
            pollers.add(new Poller(dp, directory));
        }

        if (dp.parallelDirectoryPollingEnabled) {

            /**
             * ExecutorService是一个接口类型，通过java.util.concurrent.Executors 的静态方法来创建
             * newCachedThreadPool 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程
             */
            executor = Executors.newCachedThreadPool();
        } else {
            /**
             *  newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，
             *  保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
             */
            executor = Executors.newSingleThreadExecutor();
        }
    }

    /**
     * 此方法由ExecutorService定期调用
     * This method is periodically called by the {@link ExecutorService}.
     *
     * ScheduledRunnable是Runnable接口的实现类，实现了run方法
     */
    @Override
    public synchronized void run() {
        try {
            notifier.beforePollingCycle(new BeforePollingCycleEvent(dp));
            if (!executor.isShutdown()) {
                //线程池
                executor.invokeAll(pollers);
            }
            notifier.afterPollingCycle(new AfterPollingCycleEvent(dp));
        } catch (InterruptedException e) {
            // allow thread to exit gracefully
        } catch (Throwable t) {
            LOG.error("Unexpected error!", t);
        }
    }

    void addListener(Rdp4jListener listener) {
        notifier.addListener(listener);
    }

    void removeListener(Rdp4jListener listener) {
        notifier.removeListener(listener);
    }

    void addDirectory(PolledDirectory directory) {
        pollers.add(new Poller(dp, directory));
    }

    void removeDirectory(PolledDirectory directory) {
        pollers.remove(new Poller(dp, directory));
    }

    void shutdown() {
        executor.shutdown();
    }

    void awaitTermination() {
        Util.awaitTermination(executor);
    }

    Set<PolledDirectory> getDirectories() {
        Set<PolledDirectory> result = new HashSet<PolledDirectory>();
        for (Poller poller : pollers) {
            result.add(poller.directory);
        }
        return result;
    }

}
